package com.avo.ramcache.persist.queue;

import com.avo.ramcache.config.IEntity;
import com.avo.ramcache.orm.Accessor;
import com.avo.ramcache.persist.Element;
import com.avo.ramcache.persist.Listener;
import com.avo.ramcache.persist.Persister;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

public class QueueConsumer implements Runnable {

    private static final Logger LOGGER = LoggerFactory.getLogger(QueueConsumer.class);
    private final Persister owner;
    private final Accessor accessor;
    private final String name;
    private final Thread me;
    private final BlockingQueue<Element> queue;
    private final AtomicInteger error = new AtomicInteger();

    private volatile boolean stoped;

    public QueueConsumer(String name, Accessor accessor, BlockingQueue<Element> queue, Persister owner) {
        this.accessor = accessor;
        this.queue = queue;
        this.owner = owner;
        this.name = name;
        this.me = new Thread(this, "持久化" + name + "队列");
        me.setDaemon(true);
        me.start();
    }

    @Override
    public void run() {
        while (!stoped) {
            try {
                Element element = queue.take();
                persist(element);
            } catch (InterruptedException e) {
                this.error.getAndIncrement();
                LOGGER.error("更新队列出现InterruptedException", e);
            }
        }
    }

    private void persist(Element element) {
        IEntity<?> entity = element.getEntity();
        Class clz = element.getEntityClass();
        Listener listener;
        boolean succFlag = false;
        try {
            switch (element.getType()) {
                case SAVE:
                    if (entity.serialize()) {
                        accessor.saveOrUpdate(clz, entity);
                        succFlag = true;
                    }
                    break;
                case UPDATE:
                    if (entity.serialize()) {
                        accessor.saveOrUpdate(clz, entity);
                        succFlag = true;
                    }
                    break;
                case REMOVE:
                    accessor.remove(clz, entity.getId());
                    succFlag = true;
            }

            listener = this.owner.getListener(clz);
            if (listener != null) {
                listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), null);
            }
        } catch (RuntimeException e) {
            e.printStackTrace();
            this.error.getAndIncrement();
            LOGGER.error("实体更新队列{" + name + "}处理元素{" + element + "}时data{" + entity + "}出现异常", e);
            listener = this.owner.getListener(clz);
            if (listener != null) {
                listener.notify(element.getType(), succFlag, element.getId(), element.getEntity(), e);
            }
        } catch (Exception e) {
            this.error.getAndIncrement();
            LOGGER.error("更新队列出现未知异常", e);
        }
    }

    public void stop() {
        if (LOGGER.isDebugEnabled()) {
            LOGGER.debug("定时入库[{}]收到停止通知", name);
        }
        this.stoped = true;
    }
}
